
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <netdb.h>
#include <errno.h>
#include <stdarg.h>
#include <stdio.h>

#include "../include/core.h"
#include "../include/fastArray.h"
#include "../include/connector.h"
#include "../include/session.h"
#include "../include/mylog.h"

#include "ae.h"
#include "anet.h"
#include "zmalloc.h"

static unsigned long long nextConnectorId = 0;

static int myCreateSocket(char *err, int domain) {
    int s, on = 1;
    if ((s = socket(domain, SOCK_STREAM, 0)) == -1) {
    	mylog(LOG_WARNING, "createing socket: %s", strerror(errno));
        return ANET_ERR;
    }

    /* Make sure connection-intensive things like the redis benckmark
     * will be able to close/open sockets a zillion of times */
    if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) == -1) {
    	mylog(LOG_WARNING, "setsockopt SO_REUSEADDR: %s", strerror(errno));
        return ANET_ERR;
    }
    return s;
}

int onTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData) {
	mylog(LOG_INFO, "onTimeProc~~~");
	struct connector* con = (struct connector* )clientData;
	if (!con) {
		return AE_NOMORE;
	}
	close(con->session->cfd);
	mylog(LOG_INFO, "@@@@onTimeProc~~~");
	int ret = reConnectConnector(con);
	mylog(LOG_INFO, "~~~ RET[%d] ~~~~~", ret);
	//if (ret < 0)
	//{
	//	return con->retryDelayMs;
	//}
	//return AE_NOMORE;
	if (ret == 0) {
		return AE_NOMORE; //
	}

	if (ret == 1) {
		mylog(LOG_WARNING, "SO SO!!!!");
		return AE_NOMORE;
	//	aeDeleteFileEvent(con->eventLoop, con->session->cfd, AE_WRITABLE);
	//	close(con->session->cfd);
	}
	//con->retryDelayMs *= 2;
	return con->retryDelayMs;
}

static void onConnectionCallBack(struct connector* con) {
	if (!con || !con->session) {
		return;
	}

	struct session* session = con->session;
	if (session->callBackHandle) {
		if (session->callBackHandle->onConnectionCallBack) {
			session->callBackHandle->onConnectionCallBack(session, session->interData);
		}
	}
}

static void onFileReadEvent(struct aeEventLoop * eventLoop, int fd, void *clientData, int mask) {
	mylog(LOG_INFO, "onFileReadEvent~~~");
	struct connector* con = (struct connector* )clientData;
	if (!con || !con->session) {
		return;
	}

	struct session* session = con->session;
	int len;
	len = readSession(session);
	mylog(LOG_DEBUG, "on file read event, session id[%d], len[%d]\n", session->id, len);
	if (len <= 0) {
		if (con->openReCon == OPEN) {
			aeDeleteFileEvent(con->eventLoop, session->cfd, AE_READABLE);
			close(session->cfd);
			aeCreateTimeEvent(con->eventLoop, con->retryDelayMs, onTimeProc, con, NULL);
		} else {
			if (session->callBackHandle) {
				if (session->callBackHandle->onDisConnectionCallBack) {
					session->callBackHandle->onDisConnectionCallBack(session, session->interData);
				}
			}

			aeDeleteFileEvent(eventLoop, fd, AE_READABLE);
			destroyConnector(con);
		}
	} else {
		if (session->callBackHandle) {
			if (session->callBackHandle->onDataCallBack) {
				session->callBackHandle->onDataCallBack(session, session->interData);
			}
		}

		if (getReadableBytesBuffer(session->readBuffer) <= 0) {
			retrieveAllBuffer(session->readBuffer);
		}
	}
}

struct connector* initConnector(char* serverAddr, int serverPort) {
	struct connector* con = (struct connector* )zcalloc(sizeof(struct connector));
	if (!con) {
		mylog(LOG_WARNING, "init connector error!!!");
		return NULL;
	}

	con->callBackHandle = zcalloc(sizeof(struct IServerCallBack));
	if (!con->callBackHandle) {
		mylog(LOG_WARNING, "init connector error, call back handle error!!!");
		return NULL;
	}

	con->id = nextConnectorId++;
	con->serverAddr = zstrdup(serverAddr);
	con->serverPort = serverPort;
	con->openReCon = CLOSE;
	mylog(LOG_INFO, "new connector id[%d], connection addr[%s], connection port[%d]", con->id, con->serverAddr, con->serverPort);
	return con;
}

void destroyConnector(struct connector* con) {
	if (!con) {
		return;
	}

	mylog(LOG_INFO, "destroy connector id[%d], connection addr[%s], connection port[%d]", con->id, con->serverAddr, con->serverPort);
	if (con->serverAddr) {
		zfree(con->serverAddr);
	}

	if (con->session) {
		destroySession(con->session);
		con->session = NULL;
	}

	if (con->callBackHandle) {
		zfree(con->callBackHandle);
	}

	zfree(con);
}

int startBlockConnect(connector* con, struct core* c) {
	if (!con) {
		return -1;
	}

	con->session = initSession();
	if (!con->session) {
		return -2;
	}

	char err[256];
	int ret = anetTcpConnect(err, con->serverAddr, con->serverPort);
	if (ret == ANET_ERR) {
		mylog(LOG_WARNING,  "in startBlockConnect, anetTcpConnect error!!! serverAddr[%s], serverPort[%d]", con->serverAddr, con->serverPort);
		destroySession(con->session);
		con->session = NULL;
		return -3;
	}

	con->session->cfd = ret;
	con->session->callBackHandle = con->callBackHandle;

    anetNonBlock(NULL,  con->session->cfd);
    anetTcpNoDelay(NULL, con->session->cfd);

	dispathWorkThread(c, con->session);
	return 0;
}

void onAsyncConnection(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask){
	mylog(LOG_INFO, "onAsyncConnection!!!");
	struct connector* con = (struct connector*)clientData;

	int error = 0;
	socklen_t len = sizeof(int);
	if ( (0 == getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &len))) {
		if (0 == error){
			aeDeleteFileEvent(eventLoop, fd, AE_WRITABLE);
			onConnectionCallBack(con);
		    aeCreateFileEvent(eventLoop, fd, AE_READABLE, onFileReadEvent, con);
			mylog(LOG_INFO, "connect[%s][%d] success!!!", con->serverAddr, con->serverPort);
		} else {
			aeDeleteFileEvent(eventLoop, fd, AE_WRITABLE);
			mylog(LOG_WARNING, "connetion[%s][%d] error!!!", con->serverAddr, con->serverPort);
			mylog(LOG_WARNING, "~~reconnetion[%s][%d] error!!!", con->serverAddr, con->serverPort);
			if (con->openReCon == OPEN) {
				close(fd);
				aeCreateTimeEvent(con->eventLoop, con->retryDelayMs, onTimeProc, con, NULL);
			} else {
				destroyConnector(con);
			}
		}
	} else {
		destroyConnector(con);
	}
}

int startNonBlockConnect(struct connector* con, struct core* c){
	if (!con) {
		return -1;
	}

	con->session = initSession();
	if (!con->session) {
		return -2;
	}

	con->session->sessionType = CONNECTOR_SESSIONTYPE;
	con->session->server = con;
	con->session->conAddr = zstrdup(con->serverAddr);
	con->session->conPort = con->serverPort;
	con->session->callBackHandle = con->callBackHandle;
	con->eventLoop = c->eventLoop;
	int ret = reConnectConnector(con);
	if (ret < 0) {
		destroySession(con->session);
		con->session = NULL;
	}

	return ret;
}

int reConnectConnector(struct connector* con) {
	if (!con || !con->session || !con->eventLoop) {
		return -1;
	}

	if (con->session->sessionType != CONNECTOR_SESSIONTYPE) {
		return -2;
	}

	int s;
	struct sockaddr_in sa;
	if ( (s = myCreateSocket(NULL, AF_INET)) == AE_ERR){
		return -3;
	}

    sa.sin_family = AF_INET;
    sa.sin_port = htons(con->serverPort);
	mylog(LOG_INFO, "reconnect[%s][%d] success!!!", con->serverAddr, con->serverPort);
    if (inet_aton(con->serverAddr, &sa.sin_addr) == 0) {
        struct hostent *he;

        he = gethostbyname(con->serverAddr);
        if (he == NULL) {
        	mylog(LOG_WARNING, "can't resolve: %s", con->serverAddr);
        	close(s);
            return -4;
        }
        memcpy(&sa.sin_addr, he->h_addr, sizeof(struct in_addr));
    }

    con->session->cfd = s;
    anetNonBlock(NULL,  con->session->cfd );
    anetTcpNoDelay(NULL, con->session->cfd);

    if (connect(s, (struct sockaddr*)&sa, sizeof(sa)) == -1){
    	 if (errno == EINPROGRESS){
    		 mylog(LOG_INFO, "FD: %d", s);
    		 int ret = aeCreateFileEvent(con->eventLoop, s, AE_WRITABLE, onAsyncConnection, con);
    		 if (ret == AE_ERR) {
    			 close(s);
    			 return -11;
    		 }
    		 return 1;
    	 } else {
    		 close(s);
    		 return -6;
    	 }
    }

	onConnectionCallBack(con);
    aeCreateFileEvent(con->eventLoop, s, AE_READABLE, onFileReadEvent, con);
	return 0;
}

void stopConnect(connector* con) {
	if (!con) {
		mylog(LOG_WARNING, "stop connection!!!");
		return;
	}
	mylog(LOG_INFO, "stop connection,connection id[%d], addr[%s], port[%d]\n", con->id, con->serverAddr, con->serverPort);
	if (con->session) {
		shutdownSession(con->session);
	}

	if (con->serverAddr) {
		zfree(con->serverAddr);
	}

	zfree(con);
}

void setRetryMs(struct connector* con, int ms) {
	if (!con) {
		return;
	}
	con->openReCon = OPEN ;
	con->retryDelayMs = ms * 2;
}

// no thread safe!!!
void sendConnect(struct connector* con, char* sendBuf, int sendBufLen) {
	if (!con) {
		mylog(LOG_WARNING,  "sendConnect error!!!");
		return;
	}

	if (!con->session) {
		mylog(LOG_WARNING,  "con session error!!!");
		return;
	}

	int ret = writeSession(con->session, sendBuf, sendBufLen);
	mylog(LOG_INFO, "writSession ret[%d]", ret);
	if (ret <= 0) {
		destroySession(con->session);
	}
	return;
}
